package com.cmbyc.Kafka_Flink.main


import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, Table, UnresolvedFieldExpression}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, OldCsv, Schema}


/**
 *
 * @program: com.cmbyc.Kafka_Flink.main
 * @author: YCLW058
 * @create: 2021-05-21 10:44
 * @decsription:
 *
 * */

object TableApiTest {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //并行度设置为1
    env.setParallelism(1)
    val tenv = StreamTableEnvironment.create(env)

    //2 source
    tenv.connect(new Kafka()
    .version("universal")
      .topic("hello")
      .property("zookeeper.connect", "localhost:2181")
      .property("bootstrap.servers", "localhost:9092")
    )
      .withFormat(new Csv())
      .withSchema(new Schema()
      .field("id",DataTypes.INT())
      .field("name",DataTypes.STRING())
      .field("age",DataTypes.INT()))
      .createTemporaryTable("inputTable")

    //3 transformation
    val resultTable = tenv.sqlQuery(
      """
        |select id,name
        |from inputTable
        |where id = 3
        |""".stripMargin
    )

    //4 sink
    resultTable.toAppendStream[(Int,String)].print("sql")
    //val table = tenv.from("inputTable")
    //table.toAppendStream[(Int,String,Int)].print("")
    //5 execute
    env.execute()
  }

}
